package com.qf.device.service.impl;

import cn.hutool.core.util.RandomUtil;
import cn.hutool.crypto.SecureUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.qf.device.entity.Device;
import com.qf.device.entity.DeviceTopic;
import com.qf.device.mapper.DeviceMapper;
import com.qf.device.mapper.DeviceTopicMapper;
import com.qf.dto.DriverMqttDto;
import com.qf.dto.PageDto;
import com.qf.device.service.DeviceService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.qf.mqttUser.entity.MqttUser;
import com.qf.mqttUser.mapper.MqttUserMapper;
import com.qf.vo.PageResult;
import com.qf.vo.R;
import com.qf.vo.Result;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * <p>
 * 设备表 服务实现类
 * </p>
 *
 * @author 
 * @since 2023-10-26
 */
@Service
@Slf4j
public class DeviceServiceImpl extends ServiceImpl<DeviceMapper, Device> implements DeviceService {

    @Autowired
    private DeviceMapper deviceMapper;

    @Autowired
    private MqttUserMapper mqttUserMapper;

    @Autowired
    private DeviceTopicMapper deviceTopicMapper;

    @Override
    public PageResult<Device> pageInfo(PageDto<Device> pageDto) {

        //分页信息
        IPage<Device> page = new Page<>(pageDto.getNow(),pageDto.getSize());

        //条件查询
        QueryWrapper<Device> qw = new QueryWrapper<>();
        if (!StringUtils.isEmpty(pageDto.getWhere().getProductId())){
            qw.lambda().eq(Device::getProductId,pageDto.getWhere().getProductId());
        }
        IPage<Device> pageInfo = this.baseMapper.selectPage(page,qw);

        //封装页面结构
        PageResult<Device> pageResult = new PageResult<>(pageInfo.getTotal(),pageInfo.getPages(),pageInfo.getRecords());

        return pageResult;
    }

    @Override
    public PageResult<Device> findAll(Long productId,PageDto<Device> pageDto) {
        List<Device> deviceList = deviceMapper.findAll(productId, pageDto);
        Long count = 0L;
        if (productId == -1){
            count = deviceMapper.selectCount(new QueryWrapper<>(Device.class));
        }else{
            count = deviceMapper.selectCount(new QueryWrapper<>(Device.class).lambda().eq(Device::getProductId,productId));
        }
        Long pages = (count + pageDto.getSize() - 1) / pageDto.getSize();
        PageResult<Device> pageResult = new PageResult<>(count,pages,deviceList);
        return pageResult;
    }

    @Override
    public Result<Device> findOne(Long id) {
        if (StringUtils.isEmpty(id)){
            return new Result<>(1,"非法参数");
        }
        Device device = deviceMapper.findOne(id);
        if (StringUtils.isEmpty(device)){
            return new Result<>(1,"查询失败");
        }
        return new Result<>(0,"success",device);
    }

    @Override
    @Transactional
    public R<DriverMqttDto> auth(DriverMqttDto driverMqttDto) {
        log.info("进入");
        if (StringUtils.isEmpty(driverMqttDto)){
            return R.fail(null,"非法参数");
        }
        //校验设备
        QueryWrapper<Device> qw = new QueryWrapper<>();
        qw.lambda().eq(Device::getDeviceKey,driverMqttDto.getDeviceKey()).eq(Device::getDeviceSecret,driverMqttDto.getDeviceSecret());
        Device device = this.baseMapper.selectOne(qw);
        if (device == null){
            return R.fail(driverMqttDto,"非法设备");
        }
        log.info("[设备认证]合法设备:{}",device);

        //修改设备激活状态
        device.setActive(2);
        this.baseMapper.updateById(device);

        //获取设备对应的租户对应的emqx账号（如果没有账号，生成账号）
        QueryWrapper<MqttUser> queryWrapper = new QueryWrapper();

        queryWrapper.lambda().eq(MqttUser::getTenantId,device.getTenantId());
        MqttUser mqttUser = mqttUserMapper.selectOne(queryWrapper);
        if (mqttUser == null){
            MqttUser mq = new MqttUser();
            //给租户添加账号
            mq.setUsername(device.getDeviceKey());
            mq.setPassword(SecureUtil.md5("123456"));
            mq.setCreated(new Date());
            mq.setTenantId(device.getTenantId());
            mqttUserMapper.insert(mq);
            mqttUser = mq;
            log.info("[设备认证]给租户{}添加了账户{}",device.getTenantId(),mq);
        }

        //获取设备的topic
        QueryWrapper<DeviceTopic> deviceTopicQueryWrapper = new QueryWrapper<>();
        deviceTopicQueryWrapper.lambda().eq(DeviceTopic::getDeviceKey,device.getDeviceKey());
        List<DeviceTopic> deviceTopics = deviceTopicMapper.selectList(deviceTopicQueryWrapper);
        List<DeviceTopic> data = deviceTopics.stream().filter(topic -> topic.getTopic().contains("data")).collect(Collectors.toList());
        //租户系统订阅数据上报的topic
        try {
            subTopic(data.get(0).getTopic(),mqttUser);
            log.info("[设备认证]订阅topic[{}]成功",data.get(0).getTopic());
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }

        //封装响应数据
        driverMqttDto.setUrl("tcp://43.143.125.148:1883");
        driverMqttDto.setIp("43.143.125.148:1883");
        driverMqttDto.setPort(1883);
        driverMqttDto.setUsername(mqttUser.getUsername());
        driverMqttDto.setPassword("123456");

        return R.ok(driverMqttDto,"auth success");
    }

    private void subTopic(String topic,MqttUser mqttUser) throws MqttException {
        //实例化客户端
        String broker = "tcp://43.143.125.148:1883";
        String clientId = RandomUtil.randomString(10);
        MemoryPersistence persistence = new MemoryPersistence();
        MqttClient client = new MqttClient(broker, clientId, persistence);

        // MQTT 连接选项
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(mqttUser.getUsername());
        connOpts.setPassword("123456".toCharArray());
        // 保留会话
        connOpts.setCleanSession(true);

        // 设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {

            }

            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {

                // subscribe后得到的消息会执行到这里面
                System.out.println("接收消息主题:" + topic);
                System.out.println("接收消息Qos:" + mqttMessage.getQos());
                System.out.println("接收消息内容:" + new String(mqttMessage.getPayload()));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });

        // 建立连接
        client.connect(connOpts);

        // 订阅
        client.subscribe(topic);
    }

}
